package com.zf.flink;

import com.zf.flink.simple.UserSyncJob;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringFlinkApplication  implements CommandLineRunner {

    @Autowired
    private UserSyncJob userSyncJob;

    public static void main(String[] args) {
        SpringApplication.run(SpringFlinkApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        // 启动Flink同步作业
        new Thread(() -> {
            try {
                userSyncJob.execute();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }


/*

    public static void main(String[] args) throws Exception{
        SpringApplication.run(SpringFlinkApplication.class, args);

        // 启动 Flink 流处理作业（独立环境）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 本地测试

        env.addSource(new MySQLSource())
                .keyBy(User::getId)
                .addSink(new RedisSink());

        env.execute("MyBatis-Plus + Flink + Redis Sync Job");
    }
*/


/*


    @Bean
    public CommandLineRunner flinkJobRunner(RedisSinkService redisSink) {
        return args -> {
            StreamExecutionEnvironment env =
                    StreamExecutionEnvironment.createLocalEnvironment(2); // 2并发

            // 1. CDC Source：实时拉 MySQL
            MySqlSource<String> source = MySqlSource.<String>builder()
//                    .hostname("127.0.0.1")
                    .hostname("localhost")
                    .port(3306)
                    .databaseList("bobdb")
                    .tableList("bobdb.user")
                    .username("root")
                    .password("123456")
                    .startupOptions(StartupOptions.latest())
                    .deserializer(new JsonDebeziumDeserializationSchema())
                    .build();

            DataStream<String> stream = env.fromSource(
                    source,
                    WatermarkStrategy.noWatermarks(),
                    "mysql-cdc"
            );

            // 2. 计算（这里只是透传，你可以加窗口、聚合）
            DataStream<String> result = stream
                    .map(json -> json); // TODO 你自己的业务逻辑

            // 3. Sink：写 Redis（用 Redisson）
            result.addSink(new RedisStringSink("redis://127.0.0.1:6379"));

*//*            result.addSink(new RichSinkFunction<String>() {
                private RBatch batch;
                @Override
                public void open(Configuration parameters) {
                    batch = redisSink.newBatch();
                }
                @Override
                public void invoke(String value, Context context) {
                    batch.getBucket("user:last").setAsync(value);
                }
                @Override
                public void close() {
                    batch.execute();
                }
            });
            *//*

            env.execute("mysql-cdc-to-redis");
        };
    }*/
}
